package Storage_Hbase;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;

public class FinancialMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();

        String[] fields = line.split(",");

        String rowKey = fields[0];
        String id = fields[1];
        String loanAmnt = fields[2];
        String term = fields[3];
        String interestRate = fields[4];
        String installment = fields[5];
        String grade = fields[6];
        String subGrade = fields[7];
        String employmentTitle = fields[8];
        String employmentLength = fields[9];
        String homeOwnership = fields[10];
        String annualIncome = fields[11];
        String verificationStatus = fields[12];
        String issueDate = fields[13];
        String isDefault = fields[14];
        String purpose = fields[15];
        String postCode = fields[16];
        String delinquency_2years = fields[17];
        String ficoRangeLow = fields[18];
        String ficoRangeHigh = fields[19];
        String openAcc = fields[20];
        String pubRec = fields[21];
        String pubRecBankruptcies = fields[22];
        String revolBal = fields[23];
        String revolUtil = fields[24];
        String totalAcc = fields[25];
        String initialListStatus = fields[26];
        String applicationType = fields[27];
        String earliesCreditLine = fields[28];
        String title = fields[29];
        String policyCode = fields[30];
        String n0 = fields[31];
        String n1 = fields[32];
        String n2 = fields[33];
        String n3 = fields[34];
        String n4 = fields[35];
        String n5 = fields[36];
        String n6 = fields[37];
        String n7 = fields[38];
        String n8 = fields[39];
        String n9 = fields[40];
        String n10 = fields[41];
        String n11 = fields[42];
        String n12 = fields[43];
        String n13 = fields[44];
        String n14 = fields[45];

        ImmutableBytesWritable keyOut = new ImmutableBytesWritable(Bytes.toBytes(rowKey));

        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("id"), Bytes.toBytes(id));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("loanAmnt"), Bytes.toBytes(loanAmnt));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("term"), Bytes.toBytes(term));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("interestRate"), Bytes.toBytes(interestRate));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("installment"), Bytes.toBytes(installment));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("grade"), Bytes.toBytes(grade));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("subGrade"), Bytes.toBytes(subGrade));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("employmentTitle"), Bytes.toBytes(employmentTitle));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("employmentLength"), Bytes.toBytes(employmentLength));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("homeOwnership"), Bytes.toBytes(homeOwnership));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("annualIncome"), Bytes.toBytes(annualIncome));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("verificationStatus"), Bytes.toBytes(verificationStatus));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("issueDate"), Bytes.toBytes(issueDate));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("isDefault"), Bytes.toBytes(isDefault));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("purpose"), Bytes.toBytes(purpose));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("postCode"), Bytes.toBytes(postCode));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("regionCode"), Bytes.toBytes(postCode));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("delinquency_2years"), Bytes.toBytes(delinquency_2years));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("ficoRangeLow"), Bytes.toBytes(ficoRangeLow));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("ficoRangeHigh"), Bytes.toBytes(ficoRangeHigh));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("openAcc"), Bytes.toBytes(openAcc));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("pubRec"), Bytes.toBytes(pubRec));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("pubRecBankruptcies"), Bytes.toBytes(pubRecBankruptcies));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("revolBal"), Bytes.toBytes(revolBal));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("revolUtil"), Bytes.toBytes(revolUtil));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("totalAcc"), Bytes.toBytes(totalAcc));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("initialListStatus"), Bytes.toBytes(initialListStatus));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("applicationType"), Bytes.toBytes(applicationType));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("earliesCreditLine"), Bytes.toBytes(earliesCreditLine));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("title"), Bytes.toBytes(title));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("policyCode"), Bytes.toBytes(policyCode));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n0"), Bytes.toBytes(n0));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n1"), Bytes.toBytes(n1));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n2"), Bytes.toBytes(n2));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n3"), Bytes.toBytes(n3));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n4"), Bytes.toBytes(n4));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n5"), Bytes.toBytes(n5));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n6"), Bytes.toBytes(n6));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n7"), Bytes.toBytes(n7));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n8"), Bytes.toBytes(n8));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n9"), Bytes.toBytes(n9));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n10"), Bytes.toBytes(n10));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n11"), Bytes.toBytes(n11));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n12"), Bytes.toBytes(n12));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n13"), Bytes.toBytes(n13));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("n14"), Bytes.toBytes(n14));

        context.write(keyOut, put);



    }
}
